package com.github.xiaolyuh;

import com.github.xiaolyuh.configcenter.ConfigCenter;
import com.github.xiaolyuh.constant.CacheConstant;
import com.github.xiaolyuh.entity.RedisName;
import com.github.xiaolyuh.enums.CacheCodeEnum;
import com.github.xiaolyuh.enums.ModelEnum;
import com.github.xiaolyuh.exception.CacheException;
import com.github.xiaolyuh.exception.PromotionMonitorException;
import com.github.xiaolyuh.redis.clinet.RedisClient;
import com.github.xiaolyuh.redis.clinet.SingleRedisClient;
import com.github.xiaolyuh.util.BeanFactory;
import com.github.xiaolyuh.utils.FlushHolder;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StopWatch;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

public class RedisKeyVersionManager  {

    private static final Logger logger = LoggerFactory.getLogger(RedisKeyVersionManager.class);

    private static final String LOCK = ":l";

    private static final String VERSION = ":v";

    private static final Integer RETRY_LOCK_TIMES = 3;

    private static final Integer SHARD_COUNT = 16;

    private static RedisClient redisClient = BeanFactory.getBean(SingleRedisClient.class);

    /**
     * 覆盖版本号
     *
     * @param redisName
     * @param key
     * @param expireAtTime
     * @param version
     * @return
     */
    public static Boolean updateKeyAndVersion(RedisName redisName, String key, Integer expireAtTime, String version) throws CacheException {
        Boolean locked = Boolean.FALSE;
        try {
            if (!tryLock(redisName, key)) {
                throw new CacheException(CacheCodeEnum.FAIL.getCode(), "[redis key version] try lock fail");
            }
            locked = Boolean.TRUE;
            //缓存新增操作不对比版本号,减少redis压力
            String versionKey = getVersionKey(key);
            if (FlushHolder.add() && !redisClient.hasKey(versionKey)) {
                return Boolean.TRUE;
            }
            //分片保存key,避免查询大key堵塞redis
            Integer shard = sharding(versionKey);
            redisClient.sadd(getKeyListKey(shard), key);
            //保存版本号
            redisClient.set(versionKey, version, expireAtTime, TimeUnit.SECONDS);
            logger.info("[redis key version] set redis key version key={} value={} shard={}", key, version, shard);
            return Boolean.TRUE;
        } finally {
            if (locked) unLock(redisName, key);
        }
    }

    /**
     * 查询全量的key
     *
     * @param redisName
     * @return
     */
    public static Set<String> getKeys(RedisName redisName) {
        Set<String> keyList = Sets.newHashSet();
        for (Integer i = 0; i < SHARD_COUNT; i++) {
            keyList.addAll(redisClient.smembers(getKeyListKey(i)));
        }
        return keyList;
    }

    /**
     * 查询key
     *
     * @param redisName
     * @return
     */
    public static String getRedisVersion(RedisName redisName, String key) {
        if (!ConfigCenter.allowLocalCache()) {
            return null;
        }
        String versionKey = getVersionKey(key);
        return redisClient.get(versionKey, String.class);
    }

    /**
     * 清除过期版本号
     */
    public static void clearKeyAndVersion() {
        StopWatch watch = new StopWatch();
        watch.start();
        try {
            //定时清除过期的版本号
            for (String site : ConfigCenter.getSiteCodes()) {
                for (ModelEnum model : ModelEnum.values()) {
                    RedisKeyVersionManager.clear(new RedisName(site, model));
                }
            }
        } catch (Exception e) {
            logger.error("[flush local cache]clear key occur error", PromotionMonitorException.of(e));
        } finally {
            watch.stop();
            Double second = watch.getTotalTimeSeconds();
            if (second > 5L * 60) {
                logger.warn("[flush local cache]clear key timeout cost = {}sec", second);
            }
        }
    }

    /**
     * 批量查询key
     *
     * @param redisName
     * @return
     */
    public static Map<String, String> batchGetRedisVersion(RedisName redisName, Set<String> redisKeys) {
        Map<String, String> keyAndVersion = Maps.newHashMap();
        Integer pageSize = 200;
        List<String> redisKeysList = Lists.newArrayList(redisKeys);
        for (List<String> keys : Lists.partition(redisKeysList, pageSize)) {
            try {
                List<String> versionKeys = keys.stream().map(key -> getVersionKey(key)).collect(Collectors.toList());
                String[] array = versionKeys.toArray(new String[versionKeys.size()]);
                List<String> versions = Arrays.asList(array).stream().map(k -> redisClient.get(k, String.class)).collect(Collectors.toList());
                /*List<String> versions = redisClient.mget(array);*/
                for (int i = 0; i < keys.size(); i++) {
                    keyAndVersion.put(keys.get(i), versions.get(i));
                }
            } catch (Exception e) {
                logger.error("[redis key version] get redis key version error keys=" + keys, PromotionMonitorException.of(e));
            }
        }
        return keyAndVersion;
    }

    /**
     * 清除过期key
     *
     * @param redisName
     * @return
     */
    public static void clear(RedisName redisName) {
        Set<String> keyList = getKeys(redisName);
        for (String key : keyList) {
            String versionKey = getVersionKey(key);
            Boolean locked = Boolean.FALSE;
            try {
                if (!lock(redisName, key)) {
                    continue;
                }
                locked = Boolean.TRUE;
                if (redisClient.hasKey(versionKey)) {
                    continue;
                }
                Integer shard = sharding(versionKey);
                redisClient.srem((getKeyListKey(shard)), key);
                logger.info("[redis key version] clear redis key version, key={} shard={}", key, shard);
            } finally {
                if (locked) unLock(redisName, key);
            }
        }
    }

    /**
     * 尝试拿锁
     *
     * @param redisName
     * @param key
     * @return
     */
    public static Boolean tryLock(RedisName redisName, String key) {
        //通过仿cas的方式保证拿到锁
        for (int i = 0; i < RETRY_LOCK_TIMES; i++) {
            if (lock(redisName, key)) {
                return Boolean.TRUE;
            }
            try {
                Thread.sleep(TimeUnit.MICROSECONDS.toMillis(2));
            } catch (InterruptedException e) {
                logger.error("[redis key version] lock thread interrupted, name=" + Thread.currentThread().getName(), e);
            }
        }
        return Boolean.FALSE;
    }

    /**
     * 解锁
     *
     * @param redisName
     * @param key
     */
    public static void unLock(RedisName redisName, String key) {
        redisClient.delete(getLockKey(key));
    }

    private static Boolean lock(RedisName redisName, String key) {
        String result = redisClient.setnx(getLockKey(key), CacheConstant.REDIS_RESULT_OK, 1);
        return CacheConstant.REDIS_RESULT_OK.equals(result);
    }

    /**
     * 版本号锁对应的key
     *
     * @param key
     * @return
     */
    private static String getLockKey(String key) {
        return key + LOCK;
    }

    /**
     * 版本号对应的key
     *
     * @param key
     * @return
     */
    private static String getVersionKey(String key) {
        return key + VERSION;
    }

    /**
     * 版本号集合key
     *
     * @param shard
     * @return
     */
    private static String getKeyListKey(Integer shard) {
        return CacheConstant.REDIS_KEY_VERSION_LIST + shard;
    }

    /**
     * 数据分片
     *
     * @param key
     * @return
     */
    private static Integer sharding(String key) {
        Integer hash = hash(key);
        return hash % SHARD_COUNT;
    }

    private static Integer hash(String hashKey) {
        if (hashKey == null) {
            int random = new Random().nextInt();
            if (random < 0) {
                random = 0 - random;
            }
            hashKey = new Integer(random).toString();
        }
        int hashCode = hashKey.hashCode();
        if (hashCode < 0) {
            hashCode = positive(hashCode);
        }
        return hashCode;
    }

    private static Integer positive(final Integer value) {
        if (value == Integer.MIN_VALUE) {
            return Integer.MAX_VALUE;
        } else {
            return (value < 0) ? -value : value;
        }
    }

}
